package Demo3


import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by lenovo on 2017/12/8.
  *  利用Streaming的Transform算子实现广告点击的黑名单过滤。
  */
object Streaming_TransformScala {
  def main(args: Array[String]) {
    val conf = new SparkConf().setMaster("local[2]").setAppName("Streaming_TransformScala").set("spark.testing.memory","2147480000")
   val sc = new SparkContext(conf)
   val ssc = new StreamingContext(sc,Seconds(5))

   val blockList = Array("lisi" -> true,"zhangsan" -> false)
   val blockRDD = sc.parallelize(blockList)

    val userDStream = ssc.socketTextStream("hadoop1",2222)
   val tupleDStream = userDStream.map(lines =>{
     val userName = lines.split(" ")(1)
      (userName,lines)
    })

  val userLogDStream =  tupleDStream.transform(rdd =>{

    val userRDD = rdd.leftOuterJoin(blockRDD).filter(joinRDD =>{
      println("88:"+joinRDD._2._2.getOrElse(false))
      println("99:"+joinRDD._2._2.isEmpty)
        if(joinRDD._2._2.getOrElse(false)){
          false

        }else{
           true
        }

      }).map(x =>{
      x._2._1
    })

       userRDD
    })

    userLogDStream.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
